
import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}


//attention:this code will create a process with captured port 9092
object FlumePushWordCount
{
def main(args: Array[String]): Unit = 
{
if(args.length != 2) {
System.err.println("Usage: FlumePushWordCount <hostname> <port>")
System.exit(1)
}
val Array(hostname, port) = args
val sparkConf = new SparkConf().setMaster("local[2]").setAppName("FlumePushWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(5))

val flumeStreame=FlumeUtils.createStream(ssc,hostname, port.toInt)//如果是poll方式,那么这种地方就会带有pollstream关键词,所以这里是push方式.
//所以本代码是push方式.

flumeStreame.map(x=>new String(x.event.getBody.array()).trim).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

ssc.start()
ssc.awaitTermination()
}
}

// 这个是代码来源的连接
// https://blog.csdn.net/qq_24505127/article/details/80486254
